package com.hopu.user.era;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.era.Top10Bean;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import scala.Tuple2;

import java.io.IOException;
import java.util.List;

public class Top10Shop {

    public static void main(String[] args) {

        SparkConf conf = new SparkConf().setAppName("").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");

        JavaPairRDD<Top10Bean, Integer> top10RDD = rdd.mapToPair(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String era = json.getJSONObject("birthday").getString("year").substring(2, 3) + "0";
            String type = json.getString("event_type");
            String sid = json.getString("shop_id");
            String sname = json.getString("shop_name");
            return new Tuple2<>(new Top10Bean(era, type, sid, sname), 1);
        });


        //过滤年代
        String[] eras = {"50", "60", "70", "80", "90", "00", "10"};
        for (String era: eras) {
            tranAndPut(top10RDD, era);
        }
    }


    public static void tranAndPut(JavaPairRDD<Top10Bean, Integer> top10RDD, String era) {
        //得到每个年代对应的RDD
        JavaPairRDD<Top10Bean, Integer> eraRDD = top10RDD.filter(t -> t._1.getEra().equals(era));

        //转换
        List<Top10Bean> top10Beans = transform(eraRDD);
        top10Beans.forEach(System.out::println);
        //写入数据库
//        putToHBase(top10Beans);
    }

    public static List<Top10Bean> transform(JavaPairRDD<Top10Bean, Integer> eraRDD) {
        //过滤出每个年代对应每个操作的RDD
        JavaPairRDD<Top10Bean, Integer> view = eraRDD.filter(t -> t._1.getType().equals("view"));
        JavaPairRDD<Top10Bean, Integer> cart = eraRDD.filter(t -> t._1.getType().equals("cart"));
        JavaPairRDD<Top10Bean, Integer> parchase = eraRDD.filter(t -> t._1.getType().equals("parchase"));

        //统计某个年代，某个操作类型的数量
        JavaPairRDD<Top10Bean, Integer> viewGroup =  count(view);
        JavaPairRDD<Top10Bean, Integer> cartGroup = count(cart);
        JavaPairRDD<Top10Bean, Integer> parchaseGroup = count(parchase);

        //把浏览量、加入购物车量、购买量进行连接
        JavaPairRDD<Top10Bean, Tuple2<Tuple2<Integer, Optional<Integer>>, Optional<Integer>>> result
                = viewGroup.leftOuterJoin(cartGroup).leftOuterJoin(parchaseGroup);

        JavaPairRDD<Top10Bean, Integer> countRDD = result.mapToPair(t -> {
            Top10Bean top10 = t._1;
            Tuple2<Tuple2<Integer, Optional<Integer>>, Optional<Integer>> value = t._2;
            Tuple2<Integer, Optional<Integer>> viewAndCartTuple = value._1;
            Optional<Integer> parchaseOptional = value._2;
            int viewCount = viewAndCartTuple._1;
            int cartCount = viewAndCartTuple._2.get();
            int parchaseCount = parchaseOptional.get();
            int totalCount = viewCount + cartCount + parchaseCount;

            top10.setViewCount(viewCount);
            top10.setCartCount(cartCount);
            top10.setParchaseCount(parchaseCount);
            top10.setTotalCount(totalCount);

            return new Tuple2<>(top10, totalCount);
        });
        countRDD = countRDD.sortByKey();
//        countRDD.foreach(t -> System.out.println(t));

        List<Top10Bean> top10Beans = countRDD.map(t -> t._1).take(10);


        return top10Beans;

    }

    public static void putToHBase(List<Top10Bean> top10) {

        String[] columns = {"era", "sid", "sname", "viewCount", "cartCount", "parchaseCount", "totalCount"};

        top10.forEach(t -> {
            String[] values = {t.getEra(), t.getSid(), t.getSname(), t.getViewCount()+"", t.getCartCount()+"", +
                    t.getParchaseCount()+"", t.getTotalCount()+""};
            try {
                HBaseUtils.putOneRowToHbase("era", t.getEra(), "top10shop", columns, values);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

    }

    public static JavaPairRDD<Top10Bean, Integer> count(JavaPairRDD<Top10Bean, Integer> typeRDD) {
        JavaPairRDD<Top10Bean, Iterable<Integer>> group = typeRDD.groupByKey();
        JavaPairRDD<Top10Bean, Integer> groupRDD = group.mapToPair(t -> {
            int count = 0;
            for (int v : t._2) {
                count += v;
            }
            if (t._1.getType().equals("view")) {
                t._1.setViewCount(count);
            } else if (t._1.getType().equals("cart")) {
                t._1.setCartCount(count);
            } else if (t._1.getType().equals("parchase")) {
                t._1.setParchaseCount(count);
            } else {
                throw new Exception("操作类型异常");
            }
            return new Tuple2<>(t._1, count);
        });
        return groupRDD;
    }
}
